#include "OpHandler.h"
/*
* common function
*/
static void _send_rpc_resp(void* op)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    RpcMsg* _rpc_resp = static_cast<RpcMsg*>(_op->rpc_ex->resp_msg);
    _rpc_resp->status_code = SUCCESS;
    _rpc_resp->data_len = sizeof(struct dss_msg) + _op->dss_response->data_len;
    struct dss_msg* _resp = _op->dss_response;
    _resp->msg_type = _op->dss_request->msg_type+1;
    //printf("send response\n");
    cs->rpc_server()->RpcServiceComplete(_op->rpc_ex);
    cs->rpc_server()->FreePiece(op);
}


static void _bs_process_fault(void* op, uint16_t status)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    
    _op->dss_response->status_code = status;


    /*
    \----rpc_header(struct msg_base)----\
    .....
    uint32_t data_len; // data length from dss_header to the end
    char data[0];
    \----dss_header(struct dss_msg)----\
    .....
    uint16_t header_len; // data length from chunk_resp_header to the end
    char header[0];
    \----chunk_resp_header(struct chunk_resp, or others, like struct heartbeat_resp)------\
    .....
    uint16_t data_len; // data length of inline data
    char data[0];
    \----real inline read or write data----\
    ......
    \--------------------------------------\
    */
    //before send rpc response, please set data_len in rpc_header mentioned above
    //rpc doesn't care about the rest of *_len in your header(dss_msg, chunk_resp_header)
    //because how to decode these header must be defined in callback or service function user providing. 

    //when failed, appending nothing to "struct dss_msg" 
    _op->dss_response->data_len = 0;
    SPDK_ERRLOG("Blobstore Failed to process op: msg_type = %d, status_code = %d\n", 
    _op->dss_response->msg_type, status);

    struct chunk_req *_req = reinterpret_cast<struct chunk_req *>(_op->dss_request->data); 

    SPDK_ERRLOG("error op : len = %u, off = %u\n", _req->chunk_len, _req->chunk_off);
    //in following function, we will set data_len in rpc_header(struct msg_base)
    _send_rpc_resp(op);
    
}

/*
* chunkserver create a new chunk
*/



static void _create_sync_cb(void* op, int bserrno)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg* _resp = _op->dss_response;
    struct chunk_resp* _ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    if (bserrno) {
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return;
    }
    _resp->status_code = SUCCESS;
    //note: currently, _ck_resp->id.cs_id is spdk blob ctx(we store ctx in it when calling _blob_open_cb)
    cs->id2chunk()[_ck_resp->id.blob_id] = (struct spdk_blob*)_ck_resp->id.cs_id;
    //correct cs_id
    _ck_resp->id.cs_id = cs->cs_id();
   
    //when ok, appending "struct chunk_resp" to "struct dss_msg" 
    _resp->data_len = sizeof(struct chunk_resp);
    //printf("sync op successfully, blob_id = %ld\n", _ck_resp->id.blob_id);
   
    _send_rpc_resp(op);
}

static void _blob_resize_cb(void* op, int bserrno)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    struct dss_msg* _resp = _op->dss_response;
    struct chunk_resp* _ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    if (bserrno) {
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return;
    }
    //printf("resize op successfully\n");

    spdk_blob_sync_md((struct spdk_blob*)_ck_resp->id.cs_id, _create_sync_cb, op);
}

static void 
_blob_open_cb(void* op, struct spdk_blob* blob, int bserrno)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg* _resp = _op->dss_response;
    struct chunk_resp* _ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    if (bserrno) {
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return;
    }
    cs->free_cluster_num_ = spdk_bs_free_cluster_count(cs->bs());
    uint64_t cluster_per_chunk = cs->get_cluster_per_chunk();
    if (cs->free_cluster_num_ < cluster_per_chunk) {
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return;
    }
    //store blob context temporarily
    _ck_resp->id.cs_id = (uint64_t)blob;
    //printf("open op successfully, size = %d\n", CLUSTER_PER_CHUNK);

    spdk_blob_resize(blob, cluster_per_chunk, _blob_resize_cb, op);

}

static void 
_blob_create_cb(void* op, spdk_blob_id blobid, int bserrno)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg* _resp = _op->dss_response;
    struct chunk_resp* _ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    if (bserrno) {
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return;
    }

    _ck_resp->id.blob_id = blobid;
    _ck_resp->data_len = 0;

    
    //printf("create op successfully\n");
    //open blob to do more operation like resize
    spdk_bs_open_blob(cs->bs(), blobid, _blob_open_cb, op);
}

void 
handle_op_create(ChunkOp* op)
{
    auto cs = op->cs_self;
    if(cs->is_restart())
    {
        struct dss_msg* _resp = op->dss_response;
        struct chunk_resp* _ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
        _ck_resp->id.cs_id = cs->cs_id();
        _ck_resp->id.blob_id = cs->blobs_[(cs->reuse_blob_++)%cs->blobs_.size()];
        _ck_resp->data_len = 0;
        _resp->status_code = SUCCESS;
        _resp->data_len = sizeof(struct chunk_resp);
        _send_rpc_resp(op);
        return;
    }
    spdk_bs_create_blob(cs->bs(), _blob_create_cb, op);
    return ;
}

/*
* chunkserver delete a chunk
*/

static void 
_blob_delete_cb(void* op, int bserrno)
{
    ChunkOp* _op = static_cast<ChunkOp*>(op);
    struct dss_msg* _resp = _op->dss_response;
    struct chunk_resp* _ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    if (bserrno) {
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return;
    }
    //deleting blob doesn't append "struct chunk_resp" to "struct dss_msg"
    /*
    struct chunk_resp{
        struct chunk_id;
        uint16_t data_len;
        char data[0];
    } 
    */
    _resp->data_len = 0;
    _resp->status_code = SUCCESS;

    _send_rpc_resp(op);
}

void 
handle_op_delete(ChunkOp* op)
{
    struct dss_msg *_req = op->dss_request;
    auto cs = op->cs_self;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    if(cs->id2chunk().find(_ck_req->id.blob_id) == cs->id2chunk().end()) {
        _bs_process_fault(op, INEXISTENT_CHUNK);
        return ;
    }
    spdk_bs_delete_blob(cs->bs(), _ck_req->id.blob_id, _blob_delete_cb, op);
    return ;
}

/*
* chunkserver read a chunk
*/

static void 
_blob_read_cb(void* op, int bserrno)
{
    
    if(bserrno) {
        SPDK_ERRLOG("Blobstore Failed to process op: bserrno = %d\n",bserrno); 
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return ;
    }
    ChunkOp *_op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg *_req = _op->dss_request;
    struct dss_msg *_resp = _op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    _ck_resp->id.cs_id = _op->cs_self->cs_id();
    _ck_resp->id.blob_id = reinterpret_cast<struct chunk_req*>(_op->dss_request->data)->id.blob_id;
    //printf("blobstore read blob use : %ld us\n", to_microseconds(rdtsc()-_ck_resp->id.blob_id, 0));
   
    _resp->data_len = sizeof(struct chunk_resp) + _ck_resp->data_len;
    _resp->status_code = SUCCESS;
    //_send_rpc_resp(op);
    uint64_t read_off = (_ck_req->chunk_off & (cs->io_unit_size()-1));
    memcpy(_ck_resp->data, (void*)((uint64_t)_op->alloc_buffer + read_off), _ck_req->chunk_len);
    if(_op->alloc_buffer) spdk_dma_free(_op->alloc_buffer);
    /*
    char *p = (char*)_ck_resp->data; 
    for(int i = 0; i < _ck_req->chunk_len; i++)
    {
        std::cout << *p;
        p++;
    }
    std::cout << std::endl;
    */
    _send_rpc_resp(op);
}

void 
handle_op_read(ChunkOp* op)
{
    struct dss_msg *_req = op->dss_request;
    struct dss_msg *_resp = op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    auto cs = op->cs_self;
    
    if(cs->id2chunk().find(_ck_req->id.blob_id) == cs->id2chunk().end()) {
        _bs_process_fault(op, INEXISTENT_CHUNK);
        return ;
    }
    struct spdk_blob* cur_blob = cs->id2chunk()[_ck_req->id.blob_id];
   // _ck_resp->id.blob_id = rdtsc();
    //void* prepare_msg_header(struct rpc* self, struct msg_base* header, uint8_t rpc_type,
    //uint64_t req_addr, uint32_t req_len, uint64_t resp_addr, uint32_t resp_len, uint32_t inline_len);
    //if client pass non-zero and correct value to argument "resp_addr(client local read buffer address)" 
    //and argument "resp_len(client read data size)" 
    //server will alloc the buffer locally to send data by RDMA Write,
    //else, server need append the data to server rpc response header(in struct chunk_resp->data) 
    void* io_buffer = NULL;
    //server alloc buffer to send data by RDMA Write
    _ck_resp->data_len = _ck_req->chunk_len;
    //calculate how much io_unit_size is chunk_len/chunk_off 
    uint64_t total_len = _ck_req->chunk_len + (_ck_req->chunk_off & (cs->io_unit_size()-1));
    uint64_t io_size = (total_len + cs->io_unit_size() -1) /cs->io_unit_size();
    uint64_t io_offset = _ck_req->chunk_off / cs->io_unit_size();

    //SPDK_ERRLOG("[blob-%ld] io_size = %lu , io_offset = %lu\n", _ck_req->id.blob_id, io_size, _ck_req->chunk_off); 

    io_buffer = spdk_dma_malloc(io_size*cs->io_unit_size(), 1 << 12, NULL);
    op->alloc_buffer = io_buffer;

    spdk_blob_io_read(cur_blob, cs->io_channel(), io_buffer, io_offset, io_size, _blob_read_cb , op);

    return ;
}

/*
* chunkserver write a chunk
*/

static void
_blob_read_modify_write_cb(void* op, int bserrno)
{
    if(bserrno) {
        SPDK_ERRLOG("Blobstore Failed to process op: bserrno = %d\n",bserrno); 
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return ;
    }
   
    ChunkOp *_op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg *_req = _op->dss_request;
    struct dss_msg *_resp = _op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    if(_op->alloc_buffer) spdk_dma_free(_op->alloc_buffer);
    if(_op->read_modify_write_buffer) spdk_dma_free(_op->read_modify_write_buffer);
    _send_rpc_resp(op);
}

static void
_blob_read_modify_write(void* op, int bserrno)
{
    if(bserrno) {
        SPDK_ERRLOG("Blobstore Failed to process op: bserrno = %d\n",bserrno); 
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return ;
    }
   
    ChunkOp *_op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg *_req = _op->dss_request;
    struct dss_msg *_resp = _op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);

    if(_op->alloc_buffer)
    {
        uint64_t buf = (uint64_t)_op->alloc_buffer + _ck_req->chunk_len - (_ck_req->chunk_len & (cs->io_unit_size()-1));
        memcpy(_op->read_modify_write_buffer, (void*)buf, _ck_req->chunk_len & (cs->io_unit_size()-1));
    }
    else
    {
        uint64_t buf = (uint64_t)_ck_req->data + _ck_req->chunk_len - (_ck_req->chunk_len & (cs->io_unit_size()-1));
        memcpy(_op->read_modify_write_buffer, (void*)buf, _ck_req->chunk_len & (cs->io_unit_size()-1));
    }
    uint64_t rmw_off = _ck_req->chunk_len/cs->io_unit_size() + _ck_req->chunk_off/cs->io_unit_size();
    spdk_blob_io_write(cs->id2chunk()[_ck_req->id.blob_id], cs->io_channel(), _op->read_modify_write_buffer, rmw_off, 1, _blob_read_modify_write_cb, op);

}

static void
_blob_read_modify_write2(void* op, int bserrno)
{
    if(bserrno) {
        SPDK_ERRLOG("Blobstore Failed to process op: bserrno = %d\n",bserrno); 
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return ;
    }
   
    ChunkOp *_op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg *_req = _op->dss_request;
    struct dss_msg *_resp = _op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);

    _ck_resp->id.cs_id = _op->cs_self->cs_id();
    _ck_resp->id.blob_id = _ck_req->id.blob_id;

    //write chunk, chunk resp doesn't have inline data
    _resp->data_len = sizeof(struct chunk_resp);
    _resp->status_code = SUCCESS;

    uint64_t dest = (uint64_t)_op->read_modify_write_buffer + (_ck_req->chunk_off & (cs->io_unit_size()-1));
    memcpy((void*)dest, (void*)_ck_req->data, _ck_req->chunk_len);
    /*char *p = (char*)dest; 
    for(int i = 0; i < _ck_req->chunk_len; i++)
    {
        std::cout << *p;
        p++;
    }
    std::cout << std::endl;
    */

    uint64_t rmw_off = _ck_req->chunk_off/cs->io_unit_size();
    uint64_t total_len = (_ck_req->chunk_off & (cs->io_unit_size()-1)) + _ck_req->chunk_len;
    uint64_t rmw_size = total_len/cs->io_unit_size();
    if(total_len & (cs->io_unit_size()-1)) rmw_size+=1;
    spdk_blob_io_write(cs->id2chunk()[_ck_req->id.blob_id], cs->io_channel(), _op->read_modify_write_buffer, rmw_off, rmw_size, _blob_read_modify_write_cb, op);

}

static void 
_blob_write_cb(void* op, int bserrno)
{
    if(bserrno) {
        SPDK_ERRLOG("Blobstore Failed to process op: bserrno = %d\n",bserrno); 
        _bs_process_fault(op, BLOBSTORE_ERROR);
        return ;
    }
   
    ChunkOp *_op = static_cast<ChunkOp*>(op);
    auto cs = _op->cs_self;
    struct dss_msg *_req = _op->dss_request;
    struct dss_msg *_resp = _op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);

    _ck_resp->id.cs_id = _op->cs_self->cs_id();
    _ck_resp->id.blob_id = _ck_req->id.blob_id;

    //write chunk, chunk resp doesn't have inline data
    _resp->data_len = sizeof(struct chunk_resp);
    _resp->status_code = SUCCESS;

    if(_ck_req->chunk_len & (cs->io_unit_size()-1))
    {
        _op->read_modify_write_buffer = spdk_dma_zmalloc(cs->io_unit_size(), 0x1 << 12, NULL);
        uint64_t rmw_off = _ck_req->chunk_len/cs->io_unit_size() + _ck_req->chunk_off/cs->io_unit_size();
        spdk_blob_io_read(cs->id2chunk()[_ck_req->id.blob_id], cs->io_channel(), _op->read_modify_write_buffer, rmw_off, 1, _blob_read_modify_write , op);
    }
    else
    {
        if(_op->alloc_buffer) spdk_dma_free(_op->alloc_buffer);
        _send_rpc_resp(op);
    }
    
}

void handle_op_write(ChunkOp* op)
{
    struct dss_msg *_req = op->dss_request;
    struct dss_msg *_resp = op->dss_response;
    struct chunk_req *_ck_req = reinterpret_cast<struct chunk_req*>(_req->data); 
    struct chunk_resp *_ck_resp = reinterpret_cast<struct chunk_resp*>(_resp->data);
    auto cs = op->cs_self;

    if(cs->id2chunk().find(_ck_req->id.blob_id) == cs->id2chunk().end()) {
        SPDK_ERRLOG("inexistent blob %lu\n" ,_ck_req->id.blob_id );
        _bs_process_fault(op, INEXISTENT_CHUNK);
        return ;
    }
    struct spdk_blob* cur_blob = cs->id2chunk()[_ck_req->id.blob_id];

    uint64_t io_size = 0;
    uint64_t io_offset = 0;
    void *io_buffer = 0;
 
    //if writting offset is 4kB aligned 
    if((_ck_req->chunk_off & (cs->io_unit_size()-1)) == 0)
    {
        io_offset = _ck_req->chunk_off/cs->io_unit_size();
        if(((uint64_t)io_buffer & 3) == 0)
        {
            io_buffer = _ck_req->data;
            op->alloc_buffer = 0;
        }
        else
        {
            io_buffer = spdk_dma_zmalloc(_ck_req->chunk_len, 0x1 << 12, NULL);
            op->alloc_buffer = io_buffer;
            memcpy(io_buffer, _ck_req->data, _ck_req->chunk_len);
        }
        io_size = _ck_req->chunk_len/cs->io_unit_size();
        spdk_blob_io_write(cur_blob, cs->io_channel(), io_buffer, io_offset, io_size, _blob_write_cb, op);
    }
    else
    {
        uint64_t total_len = (_ck_req->chunk_off & 0xfff) + _ck_req->chunk_len;
        uint64_t io_size = total_len/cs->io_unit_size();
        op->alloc_buffer = 0;
        if(total_len & (cs->io_unit_size()-1)) io_size+=1;
        op->read_modify_write_buffer = spdk_dma_zmalloc(io_size*cs->io_unit_size(), 0x1 << 12, NULL);
        spdk_blob_io_read(cur_blob, cs->io_channel(), op->read_modify_write_buffer, _ck_req->chunk_off/cs->io_unit_size(), io_size, _blob_read_modify_write2 , op);

    }

    return;
}

